/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.ban.ycl.wx.nrd.update.worker;

import cn.ac.iie.di.ban.ycl.wx.nrd.update.util.MongoHelper;
import cn.ac.iie.di.ban.ycl.wx.nrd.update.util.SenderPool;
import cn.ac.iie.di.datadock.rdata.exchange.client.core.session.receive.REAbstractReceiveMessageHandler;
import cn.ac.iie.di.datadock.rdata.exchange.client.exception.REConnectionException;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.ConsumePosition;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.connection.REConnection;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.REReceiveSession;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.REReceiveSessionBuilder;
import org.apache.log4j.Logger;

/**
 *
 * @author zwlin
 */
public class Receiver {

	static private final Logger logger = Logger.getLogger(Receiver.class.getName());

	REReceiveSession session;
	ReceiveHandler handler;

	public Receiver(String receiveNamesrv, String receiveTopic, String groupnamePrefix, SenderPool[] senderPools, MongoHelper[] mongos) throws REConnectionException {
		handler = new ReceiveHandler(mongos, senderPools);
		REConnection conn = new REConnection(receiveNamesrv);
		REReceiveSessionBuilder builder = (REReceiveSessionBuilder) conn.getReceiveSessionBuilder(receiveTopic);
		builder.setHandler(handler);
		builder.setFailureHandler(new REAbstractReceiveMessageHandler<byte[]>() {
			@Override
			public boolean handle(byte[] message) {
				logger.error("data format failed !");
				return true;
			}
		});
		if (groupnamePrefix == null || groupnamePrefix.trim().isEmpty()) {
			builder.setGroupName("iie_di_twx_ct_update_default_" + receiveTopic);
		} else {
			builder.setGroupName(groupnamePrefix.trim() + receiveTopic);
		}
		builder.setConsumPosition(ConsumePosition.CONSUME_FROM_LAST_OFFSET);
              // builder.setConsumPosition(ConsumePosition.CONSUME_FROM_FIRST_OFFSET);
		builder.setConsumeThreadNum(8);
		session = (REReceiveSession) builder.build();
	}

	public void start() throws REConnectionException {
		session.start();
	}
}
